package cn.gwm.flink.streaming.function.process;

import com.aliyun.datahub.client.model.RecordEntry;
import com.aliyun.datahub.client.model.TupleRecordData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Map;

/**
 * @ClassName OriginalDataProcessFunction
 * @Author xzh
 * @Date 2023/2/16 10:42
 **/
@Slf4j
public class OriginalDataProcessFunction  extends ProcessFunction<RecordEntry, String> {

    private Map<String, OutputTag<String>> tagMap;

    public OriginalDataProcessFunction(Map<String, OutputTag<String>> tagMap){
        this.tagMap = tagMap;
    }

    @Override
    public void processElement(RecordEntry recordEntry, Context context, Collector<String> collector) throws Exception {
        String modelName = "";
        String originalData = "";
        try{
            if(recordEntry!=null){
                if(recordEntry.getRecordData()!=null){
                    if(recordEntry.getRecordData() instanceof TupleRecordData){
                        //车型名称
                        modelName = (String) ((TupleRecordData) (recordEntry.getRecordData())).getField("modelname");
                        //原始数据
                        originalData = (String) ((TupleRecordData) (recordEntry.getRecordData())).getField("original_data");
                        if (StringUtils.isNotBlank(modelName) && tagMap.get(modelName) != null && StringUtils.isNotBlank(originalData)) {
                            context.output(tagMap.get(modelName), originalData);
                        }else{
                            collector.collect(originalData);
//                            log.error("原始数据不属于开放车型，modelName={},originalData={}",modelName,originalData);
                        }
                    }else{
                        log.error("recordEntry.getRecordData()源数据不是TupleRecordData对象格式");
                    }
                }else{
                    log.error("recordEntry.getRecordData()源数据对象为空");
                }
            }else{
                log.error("recordEntry源数据对象为空");
            }
        }catch (Exception e){
            log.error("datahub 原始数据格式异常，modelName={},originalData={},stackTrace={}",modelName,originalData,e.getStackTrace());
            e.printStackTrace();
            throw e;
        }
    }
}